package com.atguigu.flink.state.opearatestate;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.OperatorStateStore;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

/**
 * Created by 黄凯 on 2023/6/20 0020 18:58
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * ListState:  希望使用List集合存储数据。
 * *                      特点： 所有存储的状态，会均匀分配。
 * *
 * *                      备份的状态有: a,b,c,d
 * *                      当前Map的并行度有3:
 * *                          Map1: a,d
 * *                          Map2: b
 * *                          Map3: c
 * *
 * *     UnionListState： 希望使用List集合存储数据。
 * *                       特点： 所有存储的状态，会全量分配给每一个Task。
 * *
 * *                        备份的状态有: a,b,c,d
 * *                          当前Map的并行度有3:
 * *                             Map1: a,b,c,d
 * *                             Map2: a,b,c,d
 * *                             Map3: a,b,c,d
 * *
 * *                 kafkasource(消费者): 使用了UnionListState存储 offsets。
 * *                      假设现在Kafkasource有2个并行度。2个消费者线程，同属于一个Group。
 * *                              kafkasource1 :
 * *                                      消费 : topicA - p0 - 20
 * *                                      消费 : topicA - p2 - 21
 * *                              kafkasource2 :
 * *                                       消费 : topicA - p1 - 21
 * *
 * *                       程序挂了。重启Job，会发生reblance。由kafka的broker自动分配分区。
 * *                               kafkasource1 :
 * *                                      消费 : topicA - p0 - 20
 * *                                      消费 : topicA - p1
 * *                               kafkasource2 :
 * *                                      消费 : topicA - p2 - 21
 * *                                             topicA - p1 - 21
 * *
 * *                       使用UnionListState，可以获取到消费者偏移量的全部信息。每个kafkasource的实例都可以读取到自己要消费的分区的offsets。
 * *
 */
public class Flink03_UnionListState {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        //备份的配置信息
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();

        //开启了checkpoint（备份），每间隔2s自动备份，程序就会无限次重启
        env.enableCheckpointing(2000);

        env
                .socketTextStream("127.0.0.1", 8888)
                .map(new MyMapFunction())
                .addSink(new SinkFunction<String>() {
                    @Override
                    public void invoke(String value, Context context) throws Exception {
                        if (value.contains("x")) {
                            //人为手动抛出一个异常，为了模拟故障的情况
                            throw new RuntimeException("出异常了!");
                        }
                        System.out.println(value);
                    }
                });

        env.execute();

    }

    /*
       功能： 实现字符串的累积打印

           Flink提供的状态就是拥有自动备份，自动恢复功能的集合。
    */
    public static class MyMapFunction implements MapFunction<String, String>, CheckpointedFunction {


        //当集合用，本质是ManagerState中的OprarateState
        ListState<String> listState;

        @Override
        public String map(String value) throws Exception {
            //存
            listState.add(value);
            //取
            return listState.get().toString();
        }

        /*
           周期性(默认200ms，可以调整)将状态以快照的形式进行备份
        */
        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
            System.out.println("MyMapFunction.snapshotState");
            //备份是自动进行的，无需进行任何手动操作
        }

        /*
           初始化： 在第一次启动或任务失败重启后执行，从之前的备份中找到状态，重新赋值。

           FunctionInitializationContext context: 程序的运行环境，或上下文，可以从中获取很多额外的信息。
        */
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {

            System.out.println("MyMapFunction.initializeState");

            //找到备份状态的存储设备
            OperatorStateStore operatorStateStore = context.getOperatorStateStore();

            //从备份中找到之前备份的变量
            listState = operatorStateStore.getUnionListState(new ListStateDescriptor<>("list1", String.class));

        }
    }

}
